package Source;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import transformation.Access;

/**
 * Create by zhh on 2022/9/6.
 */
public class sourceApp {
    public static void main(String[] args) throws Exception{
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //readText(env);
        //socketTest(env);
        //richTest(env);
        mysql(env);
        env.execute();
    }
    public static void readText(StreamExecutionEnvironment env){
        DataStreamSource<String> source = env.readTextFile("data/access.log");
        System.out.println(source.getParallelism());
        source.print();

    }
    public static void socketTest(StreamExecutionEnvironment env){
        DataStreamSource<String> source = env.socketTextStream("192.168.32.222", 9527);
        System.out.println(source.getParallelism());
        source.print();
    }
    public static void richTest(StreamExecutionEnvironment env){
//        DataStreamSource<Access> source = env.addSource(new AccessSource());
//                //.setParallelism(2);只能支持单并行度
        DataStreamSource<Access> source = env.addSource(new AccessSourceV2())
        .setParallelism(2);//只能支持单并行度
        System.out.println(source.getParallelism());
        source.print();
    }
    public static void mysql(StreamExecutionEnvironment env){
        DataStreamSource<Student> source = env.addSource(new MySqlSource());//单并行度

        System.out.println(source.getParallelism());
        source.print();
    }
}
